package com.qf.kafkaconsumer.consumer;

import com.alibaba.fastjson.JSON;
import com.qf.kafkaconsumer.service.IUserService;
import com.qf.kafkapojo.pojo.User;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author q
 * @version V1.0
 * @date 2021/12/25
 **/
@Component
public class UserConsumer {

    @Autowired
    private IUserService userService;

    private final String TOPIC_NAME = "my-cluster-topic";

    /**
     * 消费者接收消息并处理
     */
    @KafkaListener(topics = TOPIC_NAME, groupId = "my-group-2")
    public void process(ConsumerRecord<String,String> record, Acknowledgment ack){
        //处理消息
        String key = record.key();
        Object value = record.value();
        System.out.println(String.format("收到的消息：key:%s, value:%s, partition:%s",key,value,record.partition()));
        if (record.key().startsWith("user")){
            User user = JSON.parseObject(record.value(), User.class);
            System.out.println("新增的user -> " + user);
            Integer row = userService.createUser(user);
            System.out.println(row == 1 ? "新增成功~ 用户id为 -> " + user.getUserId() : "新增失败");
        }
        //手动提交
        ack.acknowledge();
    }

}
